package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.article.config.stream.KafkaStreamListener;
import com.heima.common.constants.message.HotArticleConstants;
import com.heima.model.article.pojo.ArticleVisitStreamMess;
import com.heima.model.behavior.pojo.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import net.sf.jsqlparser.expression.LongValue;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.jsoup.Connection;
import org.springframework.stereotype.Component;

import java.time.Duration;

/**
 * Created by ZYJ on 2021/6/10 14:42
 */
@Component
@Slf4j
public class HotArticleStreamHandler implements KafkaStreamListener<KStream<String,String>> {
    // 接收消息 topic
    @Override
    public String listenerTopic() {
        return HotArticleConstants.HOTARTICLE_SCORE_INPUT_TOPIC;
    }
    // 流式处理成功后 发送消息 topic
    @Override
    public String sendTopic() {
        return HotArticleConstants.HOTARTICLE_INCR_HANDLE_OUPUT_TOPIC;
    }

    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {
        return stream.map((key,value)->{
            log.info("流式处理被触发 接收到的消息内容为"+value);
            UpdateArticleMess updateArticleMess = JSON.parseObject(value, UpdateArticleMess.class);
            String newKey = updateArticleMess.getArticleId()+":"+updateArticleMess.getType().name();
            String newValue = updateArticleMess.getAdd().toString();

            return new KeyValue<>(newKey,newValue);
        }).groupByKey() //按照key进行统计
          .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
          .aggregate(()->"0",(k,v,preR)->{
              return String.valueOf(Integer.valueOf(v)+Integer.valueOf(preR));
          }).toStream()
                .map((key,value)->{
                    String articleIdAndType = key.key();
                    Integer result = Integer.valueOf(value);
                    String mess = formatObj(articleIdAndType,result);
                    return new KeyValue<>(key.key(),mess);
                });
    }
    /**
     * 根据key的值来封装处理成功后的各个行为下的分值
     * @param articleIdAndType  mess.getArticleId()+":"+mess.getType()
     * @param result long类型  count之后的数值
     * @return
     */
    private String formatObj(String articleIdAndType, Integer result) {
        String[] split = articleIdAndType.split(":");
        String articleId = split[0];
        String typeName = split[1];

        ArticleVisitStreamMess streamMess = new ArticleVisitStreamMess();
        streamMess.setArticleId(Long.valueOf(articleId));
        if (UpdateArticleMess.UpdateArticleType.LIKES.name().equals(typeName)) {
            streamMess.setLike(result );
        }
        if (UpdateArticleMess.UpdateArticleType.VIEWS.name().equals(typeName)) {
            streamMess.setView(  result );
        }
        if (UpdateArticleMess.UpdateArticleType.COMMENT.name().equals(typeName)) {
            streamMess.setComment( result);
        }
        if (UpdateArticleMess.UpdateArticleType.COLLECTION.name().equals(typeName)) {
            streamMess.setCollect(result);
        }

        return JSON.toJSONString(streamMess);
    }
}
